package com.google.code.p.restrpc.message;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.google.code.p.restrpc.core.thread.NamedThreadFactory;

/**
 * @author dz
 */
public class DefaultMessageProcessor implements MessageProcessor, MessageQueue {
    
    private static final int DEFAULT_LAZY_TIME = 60 * 5;
    
    private ExecutorService executorService;
    
    private ScheduledExecutorService scheduledExecutorService;
    
    private ConcurrentHashMap<String, MessageAction> actionMap = new ConcurrentHashMap<String, MessageAction>();
    
    private int delay = -1;
    
    /**
     * 
     */
    public DefaultMessageProcessor() {
        this(-1);
    }
    
    /**
     * @param delay
     */
    public DefaultMessageProcessor(int delay) {
        this.executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
                                                                   .availableProcessors(),
                                                            new NamedThreadFactory("restrpc-message-sender",
                                                                                   true));
        this.scheduledExecutorService = Executors.newScheduledThreadPool(Runtime.getRuntime()
                                                                                .availableProcessors(),
                                                                         new NamedThreadFactory("restrpc-message-scheduled-sender",
                                                                                                true));
        this.delay = delay;
    }
    
    /**
     * @param poolSize
     * @param delay
     */
    public DefaultMessageProcessor(int poolSize, int delay) {
        this.executorService = Executors.newFixedThreadPool(poolSize);
        this.scheduledExecutorService = Executors.newScheduledThreadPool(poolSize);
        this.delay = delay;
    }
    
    /**
     * @param name
     * @param messageAction
     */
    public void put(String name, MessageAction messageAction) {
        actionMap.put(name, messageAction);
    }
    
    /**
     * @param clazz
     * @param messageAction
     */
    public void put(Class clazz, MessageAction messageAction) {
        actionMap.put(clazz.getName(), messageAction);
    }
    
    /**
     * @param name
     * @return
     */
    public MessageAction remove(String name) {
        return actionMap.remove(name);
    }
    
    /**
     * @param clazz
     * @return
     */
    public MessageAction remove(Class clazz) {
        return actionMap.remove(clazz.getName());
    }
    
    /**
     * 
     */
    public void stop() {
        if (!executorService.isShutdown()) {
            executorService.shutdown();
        }
    }
    
    /**
     * @return
     */
    public MessageQueue getDataPipe() {
        return this;
    }
    
    /**
     * 
     */
    public void addMessage(final Object data) {
        executorService.execute(new Runnable() {
            
            public void run() {
                MessageAction action = actionMap.get(data.getClass().getName());
                if (null == action) {
                    return;
                }
                
                action.execute(new MessageContext() {
                    
                    public Object getMessage() {
                        return data;
                    }
                    
                    public MessageQueue getQueue() {
                        return DefaultMessageProcessor.this;
                    }
                    
                });
            }
            
        });
    }
    
    /**
     * 
     */
    public void addLazyMessage(final Object data) {
        if (delay <= 0) {
            addLazyMessage(data, DEFAULT_LAZY_TIME);
            return;
        }
        addLazyMessage(data, delay);
    }
    
    /**
     * 
     */
    public void addLazyMessage(final Object data, int delay) {
        scheduledExecutorService.schedule(new Runnable() {
            
            public void run() {
                MessageAction action = actionMap.get(data.getClass().getName());
                if (null == action) {
                    return;
                }
                
                action.execute(new MessageContext() {
                    
                    public Object getMessage() {
                        return data;
                    }
                    
                    public MessageQueue getQueue() {
                        return DefaultMessageProcessor.this;
                    }
                    
                });
            }
            
        },
                                          delay,
                                          TimeUnit.SECONDS);
    }
    
}
